package com.lemon.cloud.curator.lock;

import com.lemon.cloud.lock.AbstractDistributedLock;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.springframework.util.Assert;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author lemon
 * @version 1.0
 * @description: TODO
 * @date Create by lemon on 2019-09-12 13:01
 */
@Slf4j
public class CuratorDistributedLock extends AbstractDistributedLock {
    @Setter
    @Getter
    private CuratorFramework curatorFramework;

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private final static String ROOT_PATH_LOCK = "/rootlock";

    /**
     * @param key
     * @param expireTime
     * @param expireTimeUnit
     * @return boolean
     * @description
     * @author lemon
     * @date 2019-09-08 20:26
     */
    @Override
    protected boolean acquire(String key, long expireTime, TimeUnit expireTimeUnit) {
        try {
            curatorFramework
                    .create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(String.format("%s/%s", CuratorDistributedLock.ROOT_PATH_LOCK, key));
            return true;
        } catch (Exception e) {
            log.error("Obtaining an exception for the curatorFramework distributed lock", e);

            try {
                if (countDownLatch.getCount() <= 0) {
                    countDownLatch = new CountDownLatch(1);
                }

                countDownLatch.await();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }

        return false;
    }

    /**
     * @param key key值
     * @return void
     * @description 释放锁
     * @author lemon
     * @date 2019-09-04 12:49
     */
    @Override
    public boolean unlock(String key) {
        key = String.format("%s/%s", CuratorDistributedLock.ROOT_PATH_LOCK, key);

        try {
            if (curatorFramework.checkExists().forPath(key) != null) {
                curatorFramework.delete().forPath(key);
            }
        } catch (Exception e) {
            log.error("Release curatorFramework lock occured an exception", e);
            return false;
        }

        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.state(this.getCuratorFramework() != null, "CuratorFramework is required");

        this.curatorFramework = this.curatorFramework.usingNamespace("lock-namespace");

        try {
            if (this.curatorFramework.checkExists().forPath(ROOT_PATH_LOCK) == null) {
                this.curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(ROOT_PATH_LOCK);
            }

            addWatcher(ROOT_PATH_LOCK);
            log.info("Zookeeper distributed lock root path {} watcher 事件创建成功", ROOT_PATH_LOCK);
        } catch (Exception e) {
            log.error(String.format("Connect zookeeper fail: {}", e.getMessage()), e);
        }
    }

    private void addWatcher(String path) throws Exception {
        String keyPath;

        if (path.equals(ROOT_PATH_LOCK)) {
            keyPath = path;
        } else {
            keyPath = String.format("%s/%s", CuratorDistributedLock.ROOT_PATH_LOCK, path);
        }

        final PathChildrenCache cache = new PathChildrenCache(this.curatorFramework, keyPath, false);

        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((client, event) -> {
            if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                String oldPath = event.getData().getPath();
                log.info("Success to release lock for path:{}", oldPath);

                if (oldPath.contains(path)) {
                    // 释放计数器，让当前的请求获取锁
                    this.countDownLatch.countDown();
                }
            }
        });
    }
}
